/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.common;

import cn.sunny360.core.handler.msg.MsgAction;
import cn.sunny360.core.service.UserService;
import cn.sunny360.core.utils.Container;
import cn.sunny360.core.utils.ParamUtils;
import cn.sunny360.core.wrapper.CacheWrapper;
import cn.sunny360.core.wrapper.DBWrapper;
import cn.sunny360.messsage.MsgMessage;
import io.vertx.core.*;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;
import io.vertx.core.json.JsonObject;
import io.vertx.core.shareddata.AsyncMap;
import io.vertx.core.shareddata.SharedData;
import io.vertx.sqlclient.Tuple;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;

import java.util.Map;

public class WebSocketHandler {

    private static Logger logger = LoggerFactory.getLogger(WebSocketHandler.class);
    private String CHAT_URL = "/phoenix/chat------------------";

    private Vertx vertx;
    private EventBus eventBus = null;
    private MsgActionMapping msgActionMapping;
    private MessageInterceptorChain messageInterceptorChain;
    private UserService userService;

    public WebSocketHandler(Vertx vertx) {
        this.vertx = vertx;
        eventBus = vertx.eventBus();
        msgActionMapping = Container.getComponent(MsgActionMapping.class);
        userService = Container.getComponent(UserService.class);
    }

    public void setMessageInterceptorChain(MessageInterceptorChain messageInterceptorChain) {
        this.messageInterceptorChain = messageInterceptorChain;
    }

    public void setCHAT_URL(String CHAT_URL) {
        this.CHAT_URL = CHAT_URL;
    }

    public void webSocketHandler(ServerWebSocket webSocket) {
        logger.info("webSocket Connected！");

        // 接收客户端连接
        if (!webSocket.path().equals(CHAT_URL)) {
            logger.info("websocket路径[" + webSocket.path() + "]非法！拒绝连接！");
            webSocket.reject();
            return;
        }
        String query = webSocket.query();
        if (StringUtils.isEmpty(query)) {
            logger.info("websocket路径query非法！拒绝连接！");
            webSocket.reject();
            return;
        }
        Map<String, String> analysis = ParamUtils.analysis(query);
        String token = analysis.get(ParamUtils.USER_TOEKN);
        if(StringUtils.isEmpty(token)){
            logger.info("token为空！拒绝连接！");
            webSocket.reject();
            return;
        }

        Promise<Boolean> queryPromise = Promise.promise();
        userService.queryUserByToken(token, queryPromise);
        queryPromise.future().onFailure(e -> {
                    queryPromise.handle(Future.failedFuture(e));
                    webSocket.close();
                })
                .onSuccess(optional -> {
                    if (!optional) {
                        logger.info("token非法！关闭连接！" + token);
                        webSocket.close();
                    }
                });


        // websocket接收到消息就会调用此方法
        webSocket.handler(buffer -> {
            routeMessage(webSocket, buffer);
        });
        // 当连接关闭后就会调用此方法
        webSocket.closeHandler(v -> {
            closeConnection(webSocket);
        });
        // WebSocket异常处理器
        webSocket.exceptionHandler(this::wsExceptionHandler);
    }

    /**
     * 消息路由，处理不同类型的消息
     *
     * @param buffer 消息内容
     */
    private void routeMessage(ServerWebSocket webSocket, Buffer buffer) {
        if (messageInterceptorChain != null) {
            messageInterceptorChain.handle(webSocket, buffer)
                    .future()
                    .onSuccess(b -> {
                        handleMessage(webSocket, b);
                    })
                    .onFailure(e -> {
                        if (e instanceof MessageInterceptorException) {
                            //拦截器正常拦截
                            logger.info("拦截器拦截", e);
                        } else {
                            //拦截器中发生异常
                            logger.error("拦截器发生异常", e);
                        }
                    });
        } else {
            handleMessage(webSocket, buffer);
        }

    }

    private void handleMessage(ServerWebSocket webSocket, Buffer buffer) {
        MsgMessage message = MsgMessage.newInstance();
        message.parseByteArray(buffer.getBytes());
        String type = message.getType();
        MsgAction action = msgActionMapping.getAction(type);
        if (action != null) {
            action.handler(vertx, webSocket, message)
                    .future()
                    .onSuccess(r -> {
                        logger.debug("message handle success:" + r);
                    })
                    .onFailure(e -> {
                        StringBuilder sb = new StringBuilder()
                                .append("\n\t1.message handle error.\n")
                                .append("\t2.error action:")
                                .append(action.getClass())
                                .append("\n")
                                .append("\t3.error message:\n")
                                .append(message);
                        logger.error(sb.toString(), e);
                    });
        } else {
            logger.error("unsupported message type[" + type + "]", new Exception(message.toJson()));
        }
    }

    private void closeConnection(ServerWebSocket webSocket) {
        // 取Websocket和token之间的对应关系
        SharedData sd = vertx.sharedData();
        Future<AsyncMap<String, String>> mapFuture = null;
        if (vertx.isClustered()) {
            mapFuture = sd.getClusterWideMap("ws_token_map");
        } else {
            mapFuture = sd.getAsyncMap("ws_token_map");
        }
        mapFuture.onSuccess(asyncMap -> {
            asyncMap.get(webSocket.binaryHandlerID(), h -> {
                if (h.succeeded()) {

                    String empty = h.result();
                    if (StringUtils.isBlank(empty)) {
                        logger.debug("连接前拒绝,不用操作了");
                        return;
                    }

                    String token = h.result().toString();
                    logger.debug("远程连接关闭，token为：" + token);


                    //删除Map中token和Websocket的对应关系
                    asyncMap.remove(webSocket.binaryHandlerID(), rem -> {
                        if (rem.succeeded()) {
                            logger.debug("删除Map中token和Websocket的对应关系");
                        } else {
                            logger.error("关闭websocket连接异常", rem.cause());
                        }
                    });
                    //创建一个关闭连接的消息
                    MsgMessage message = MsgMessage.newInstance()
                            .setType("close")
                            .setTime(new DateTime().toString("yyyy-MM-dd HH:mm:ss"));
                    //关闭连接
                    eventBus.send(token, Buffer.buffer(message.toByteArray()));

                } else {
                    logger.error("关闭websocket连接异常", h.cause());
                }
            });
        }).onFailure(e -> {
            logger.error("关闭websocket连接异常", e);
        });
    }

    private void wsExceptionHandler(Throwable e) {
        logger.error("WebSocket服务异常", e);
    }


}
